#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <unistd.h>
#include <signal.h>
#include <sys/types.h>
#include <sys/wait.h>
#include "MQTTAsync.h"
#include "MQTTClient.h"

#define CLIENTIDA   "threadA"
#define CLIENTIDB  	"threadB"

#define TOPICAPUB   "threadAPub"
#define TOPICASUB   "threadBPub"

#define TOPICBPUB   "threadBPub"
#define TOPICBSUB   "threadAPub"

#define PAYLOADA    "Messege from threadA"
#define PAYLOADB 	"Messege from threadB"

#define QOS         1
#define TIMEOUT     10000L

#define USERNAME "admin"
#define PASSWORD "password"

int disc_finished = 0;
int subscribed = 0;
int finished = 0;

int receivedA = 0;
int receivedB = 0;

int exit_flag = 1;

void exitHandler(int signum)
{
	exit_flag = 0;
}

void onConnectA(void* context, MQTTAsync_successData* response);
void onConnectFailureA(void* context, MQTTAsync_failureData* response);
void connlostA(void *context, char *cause);
void onDisconnectFailureA(void* context, MQTTAsync_failureData* response);
void onDisconnectA(void* context, MQTTAsync_successData* response);
void onSubscribeA(void* context, MQTTAsync_successData* response);
int msgarrvdA(void *context, char *topicName, int topicLen, MQTTAsync_message *message);

void onConnectB(void* context, MQTTAsync_successData* response);
void onConnectFailureB(void* context, MQTTAsync_failureData* response);
void connlostB(void *context, char *cause);
void onDisconnectFailureB(void* context, MQTTAsync_failureData* response);
void onDisconnectB(void* context, MQTTAsync_successData* response);
void onSubscribeB(void* context, MQTTAsync_successData* response);
int msgarrvdB(void *context, char *topicName, int topicLen, MQTTAsync_message *message);

int threadA(MQTTAsync_connectOptions * conn_opts,MQTTAsync_disconnectOptions *disc_opts,char *serverURI);
int threadB(MQTTAsync_connectOptions * conn_opts,MQTTAsync_disconnectOptions *disc_opts,char *serverURI);

int main(int argc,char *argv[])
{
	int pid;
	MQTTAsync_connectOptions conn_opts = MQTTAsync_connectOptions_initializer;
	MQTTAsync_disconnectOptions disc_opts = MQTTAsync_disconnectOptions_initializer;

	conn_opts.username = USERNAME;
	conn_opts.password = PASSWORD;

	signal(SIGINT,exitHandler);

	if((pid = fork()) == 0)
		threadB(&conn_opts,&disc_opts,argv[1]);
	else
		//wait(NULL);
		threadA(&conn_opts,&disc_opts,argv[1]);
	
	return 0;
}

void connlostA(void *context, char *cause)
{
	MQTTAsync client = (MQTTAsync)context;
	MQTTAsync_connectOptions conn_opts = MQTTAsync_connectOptions_initializer;
	int rc;

	conn_opts.username = USERNAME;
	conn_opts.password = PASSWORD;

	printf("\nThread A Connection lost\n");
	if (cause)
		printf("     cause: %s\n", cause);

	printf("Reconnecting\n");
	conn_opts.keepAliveInterval = 20;
	conn_opts.cleansession = 1;
	conn_opts.onSuccess = onConnectA;
	conn_opts.onFailure = onConnectFailureA;
	if ((rc = MQTTAsync_connect(client, &conn_opts)) != MQTTASYNC_SUCCESS)
	{
		printf("Failed to start connect, return code %d\n", rc);
		finished = 1;
	}
}

void onDisconnectFailureA(void* context, MQTTAsync_failureData* response)
{
	printf("Thread A disconnect failed, rc %d\n", response->code);
	disc_finished = 1;
}

void onDisconnectA(void* context, MQTTAsync_successData* response)
{
	printf("Thread A successful disconnection\n");
	disc_finished = 1;
}

void onSubscribeA(void* context, MQTTAsync_successData* response)
{
	printf("Thread A subscribe succeeded\n");
	
	subscribed = 1;
}

void onSubscribeFailureA(void* context, MQTTAsync_failureData* response)
{
	printf("Thread A subscribe failed, rc %d\n", response->code);
	finished = 1;
}

void onConnectFailureA(void* context, MQTTAsync_failureData* response)
{
	printf("Thread A connect failed, rc %d\n", response->code);
	finished = 1;
}

void onConnectA(void* context, MQTTAsync_successData* response)
{
	MQTTAsync client = (MQTTAsync)context;
	MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer;
	int rc;

	printf("Successful connection\n");

	printf("Subscribing to topic %s\nfor client %s using QoS%d\n\n"
			"Press Q<Enter> to quit\n\n", TOPICASUB, CLIENTIDA, QOS);
	opts.onSuccess = onSubscribeA;
	opts.onFailure = onSubscribeFailureA;
	opts.context = client;
	if ((rc = MQTTAsync_subscribe(client, TOPICASUB, QOS, &opts)) != MQTTASYNC_SUCCESS)
	{
		printf("Failed to start subscribe, return code %d\n", rc);
		finished = 1;
	}
}

int msgarrvdA(void *context, char *topicName, int topicLen, MQTTAsync_message *message)
{
    printf("Thread A message arrived\n");
    printf("     topic: %s\n", topicName);
    printf("   message: %.*s\n\n", message->payloadlen, (char*)message->payload);
    MQTTAsync_freeMessage(&message);
    MQTTAsync_free(topicName);
	receivedA = 1;

    return 1;
}

int threadA(MQTTAsync_connectOptions * conn_opts,MQTTAsync_disconnectOptions *disc_opts,char *serverURI)
{
	static int sendonce = 1;
	MQTTAsync client;
	MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer;
	MQTTAsync_message pubmsg = MQTTAsync_message_initializer;
	int rc;
	int ch;

	if ((rc = MQTTAsync_create(&client,serverURI, "threadA", MQTTCLIENT_PERSISTENCE_NONE, NULL))
			!= MQTTASYNC_SUCCESS)
	{
		printf("Failed to create client, return code %d\n", rc);
		rc = EXIT_FAILURE;
		goto exit;
	}

	if ((rc = MQTTAsync_setCallbacks(client, client, connlostA, msgarrvdA, NULL)) != MQTTASYNC_SUCCESS)
	{
		printf("Failed to set callbacks, return code %d\n", rc);
		rc = EXIT_FAILURE;
		goto destroy_exit;
	}

	conn_opts->keepAliveInterval = 20;
	conn_opts->cleansession = 1;
	conn_opts->onSuccess = onConnectA;
	conn_opts->onFailure = onConnectFailureA;
	conn_opts->context = client;
	if ((rc = MQTTAsync_connect(client, conn_opts)) != MQTTASYNC_SUCCESS)
	{
		printf("Failed to start connect, return code %d\n", rc);
		rc = EXIT_FAILURE;
		goto destroy_exit;
	}

	while (!subscribed && !finished)
		usleep(10000L);

	if (finished)
		goto exit;

	opts.onSuccess = NULL;
	opts.onFailure = NULL;
	opts.context = client;
	pubmsg.payload = PAYLOADA;
	pubmsg.payloadlen = (int)strlen(PAYLOADA);
	pubmsg.qos = QOS;
	pubmsg.retained = 0;

	do 
	{
		if(sendonce == 1) //发送一个消息激活循环
		{
			pubmsg.retained = 1;
			if ((rc = MQTTAsync_sendMessage(client, TOPICAPUB, &pubmsg, &opts)) != MQTTASYNC_SUCCESS)
			{
				printf("Failed to send first message, return code %d\n", rc);
				exit(EXIT_FAILURE);
			}
			pubmsg.retained = 0;
			sendonce = 0;
		}

		if(receivedA == 1)
		{
			sleep(5);
			if ((rc = MQTTAsync_sendMessage(client, TOPICAPUB, &pubmsg, &opts)) != MQTTASYNC_SUCCESS)
			{
				printf("Failed to start sendMessage, return code %d\n", rc);
				exit(EXIT_FAILURE);
			}
			receivedA = 0;
		}
		//ch = getchar();
	} while (exit_flag);

	disc_opts->onSuccess = onDisconnectA;
	disc_opts->onFailure = onDisconnectFailureA;
	if ((rc = MQTTAsync_disconnect(client, disc_opts)) != MQTTASYNC_SUCCESS)
	{
		printf("Failed to start disconnect, return code %d\n", rc);
		rc = EXIT_FAILURE;
		goto destroy_exit;
	}	
	while (!disc_finished)
 	{
			usleep(10000L);
 	}
destroy_exit:
	MQTTAsync_destroy(&client);
exit:
    return rc;
}


void connlostB(void *context, char *cause)
{
	MQTTAsync client = (MQTTAsync)context;
	MQTTAsync_connectOptions conn_opts = MQTTAsync_connectOptions_initializer;
	int rc;

	conn_opts.username = USERNAME;
	conn_opts.password = PASSWORD;

	printf("\nThread B Connection lost\n");
	if (cause)
		printf("     cause: %s\n", cause);

	printf("Reconnecting\n");
	conn_opts.keepAliveInterval = 20;
	conn_opts.cleansession = 1;
	conn_opts.onSuccess = onConnectB;
	conn_opts.onFailure = onConnectFailureB;
	if ((rc = MQTTAsync_connect(client, &conn_opts)) != MQTTASYNC_SUCCESS)
	{
		printf("Failed to start connect, return code %d\n", rc);
		finished = 1;
	}
}

void onDisconnectFailureB(void* context, MQTTAsync_failureData* response)
{
	printf("Thread B disconnect failed, rc %d\n", response->code);
	disc_finished = 1;
}

void onDisconnectB(void* context, MQTTAsync_successData* response)
{
	printf("Thread B successful disconnection\n");
	disc_finished = 1;
}

void onSubscribeB(void* context, MQTTAsync_successData* response)
{
	printf("Thread B subscribe succeeded\n");
	
	subscribed = 1;
}

void onSubscribeFailureB(void* context, MQTTAsync_failureData* response)
{
	printf("Thread B subscribe failed, rc %d\n", response->code);
	finished = 1;
}

void onConnectFailureB(void* context, MQTTAsync_failureData* response)
{
	printf("Thread B connect failed, rc %d\n", response->code);
	finished = 1;
}

void onConnectB(void* context, MQTTAsync_successData* response)
{
	MQTTAsync client = (MQTTAsync)context;
	MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer;
	int rc;

	printf("Successful connection\n");

	printf("Subscribing to topic %s\nfor client %s using QoS%d\n\n"
			"Press Q<Enter> to quit\n\n", TOPICBSUB, CLIENTIDB, QOS);
	opts.onSuccess = onSubscribeB;
	opts.onFailure = onSubscribeFailureB;
	opts.context = client;
	if ((rc = MQTTAsync_subscribe(client, TOPICBSUB, QOS, &opts)) != MQTTASYNC_SUCCESS)
	{
		printf("Failed to start subscribe, return code %d\n", rc);
		finished = 1;
	}
}

int msgarrvdB(void *context, char *topicName, int topicLen, MQTTAsync_message *message)
{
    printf("Thread B message arrived\n");
    printf("     topic: %s\n", topicName);
    printf("   message: %.*s\n\n", message->payloadlen, (char*)message->payload);
    MQTTAsync_freeMessage(&message);
    MQTTAsync_free(topicName);
	receivedB = 1;

    return 1;
}


int threadB(MQTTAsync_connectOptions * conn_opts,MQTTAsync_disconnectOptions *disc_opts,char *serverURI)
{
	MQTTAsync client;
	MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer;
	MQTTAsync_message pubmsg = MQTTAsync_message_initializer;
	int rc;
	int ch;

	if ((rc = MQTTAsync_create(&client,serverURI, "threadB", MQTTCLIENT_PERSISTENCE_NONE, NULL))
			!= MQTTASYNC_SUCCESS)
	{
		printf("Failed to create client, return code %d\n", rc);
		rc = EXIT_FAILURE;
		goto exit;
	}

	if ((rc = MQTTAsync_setCallbacks(client, client, connlostB, msgarrvdB, NULL)) != MQTTASYNC_SUCCESS)
	{
		printf("Failed to set callbacks, return code %d\n", rc);
		rc = EXIT_FAILURE;
		goto destroy_exit;
	}

	conn_opts->keepAliveInterval = 20;
	conn_opts->cleansession = 1;
	conn_opts->onSuccess = onConnectB;
	conn_opts->onFailure = onConnectFailureB;
	conn_opts->context = client;
	if ((rc = MQTTAsync_connect(client, conn_opts)) != MQTTASYNC_SUCCESS)
	{
		printf("Failed to start connect, return code %d\n", rc);
		rc = EXIT_FAILURE;
		goto destroy_exit;
	}

	while (!subscribed && !finished)
		usleep(10000L);

	if (finished)
		goto exit;

	opts.onSuccess = NULL;
	opts.onFailure = NULL;
	opts.context = client;
	pubmsg.payload = PAYLOADB;
	pubmsg.payloadlen = (int)strlen(PAYLOADB);
	pubmsg.qos = QOS;
	pubmsg.retained = 0;

	do 
	{
		if(receivedB == 1)
		{
			sleep(5);
			//alarm(60);
			//pause();
			if ((rc = MQTTAsync_sendMessage(client, TOPICBPUB, &pubmsg, &opts)) != MQTTASYNC_SUCCESS)
			{
				printf("Failed to start sendMessage, return code %d\n", rc);
				exit(EXIT_FAILURE);
			};
			receivedB = 0;
		}
		//ch = getchar();
	} while (exit_flag);

	disc_opts->onSuccess = onDisconnectB;
	disc_opts->onFailure = onDisconnectFailureB;
	if ((rc = MQTTAsync_disconnect(client, disc_opts)) != MQTTASYNC_SUCCESS)
	{
		printf("Failed to start disconnect, return code %d\n", rc);
		rc = EXIT_FAILURE;
		goto destroy_exit;
	}	
	while (!disc_finished)
 	{
			usleep(10000L);
 	}
destroy_exit:
	MQTTAsync_destroy(&client);
exit:
    return rc;
}